package core

import (
	"context"
	"fmt"
	"github.com/spf13/cast"
	"google.golang.org/grpc"
	"log"
	"net"
	"server/component"
	"server/im/action"
	"server/im/entity"
	"server/im/protocol"
	pb "server/im/rpc"
	"server/service"
	"sync"
)

var CoreHub *Hub

func NewHub(ctx context.Context) *Hub {
	CoreHub = &Hub{
		respondChan:  make(chan entity.CoreResultModel, 2),
		online:       make(map[string]entity.HubOnlineItem),
		onlineChange: make(chan entity.OnlineChangeEvent, 1024),
		gatewayMap:   sync.Map{},
		ctx:          ctx,
		exitChan:     make(chan struct{}),
	}
	return CoreHub
}

type Hub struct {
	respondChan  chan entity.CoreResultModel
	online       map[string]entity.HubOnlineItem
	onlineChange chan entity.OnlineChangeEvent
	gatewayMap   sync.Map //map[string]*Gateway
	ctx          context.Context
	innerWG      sync.WaitGroup
	exitChan     chan struct{}
}

func (o *Hub) hubInit() {
	//初始化用户
	err := service.Us.InitOnline(o.ctx)
	if err != nil {
		panic(err)
	}
	//注册事件
	o.regEvent()
}

func (o *Hub) Start(wg *sync.WaitGroup) {
	defer func() {
		fmt.Println("hub wg done")
		wg.Done()
	}()
	fmt.Println("hub starting")
	o.hubInit()
	o.innerWG.Add(2)
	go o.startListen()
	go o.startRpc()
	o.innerWG.Wait()
	fmt.Println("hub wait out")
}

func (o *Hub) Stop() {
	fmt.Println("hub will stop")
	close(o.exitChan)
}

// 开启rpc server
func (o *Hub) startRpc() {
	defer o.innerWG.Done()
	rpcPort := cast.ToString(component.ConfigModel.RpcPort)
	lis, err := net.Listen("tcp", ":"+rpcPort)
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterHubServer(s, &rpcServer{hub: o})

	go func() {
		if err := s.Serve(lis); err != nil {
			log.Fatalf("failed to rpc: %v", err)
		}
	}()

	log.Printf("hub rpc listening at %v", lis.Addr())

	//监听退出
	select {
	case <-o.exitChan:
		s.GracefulStop()
		lis.Close()
		log.Println("rpc服务已关闭")
	}
}

func (o *Hub) addGateway(gateway *Gateway) {
	o.gatewayMap.Store(gateway.gatewayId, gateway)
	gateway.run(o.ctx)
}

// 开启数据监听
func (o *Hub) startListen() {
	defer o.innerWG.Done()
LISTEN:
	for {
		select {
		case <-o.exitChan:
			fmt.Println("hub stop listen")
			break LISTEN
		//响应
		case result, ok := <-o.respondChan:
			if ok {
				if result.ResultType == entity.CoreResultTypeOne {
					value, ok := o.gatewayMap.Load(result.CoreOutModel.GatewayId)
					if ok {
						gateway, ok := value.(*Gateway)
						if !ok {
							fmt.Errorf("gateway err")
						}
						gateway.write(result.CoreOutModel)
					}
				}
				if result.ResultType == entity.CoreResultTypeAll {
					o.gatewayMap.Range(func(key, value any) bool {
						fmt.Println(key, value)
						gateway, ok := value.(*Gateway)
						if !ok {
							fmt.Errorf("gateway err")
						}
						gateway.write(result.CoreOutModel)
						return true // 返回 true 继续遍历
					})
				}
			}
		}
	}
}

// 注册事件
func (o *Hub) regEvent() {
	//登录
	component.DefaultEventBus.Subscribe(protocol.EventUserLogin, action.EventLogin)
	//登出
	component.DefaultEventBus.Subscribe(protocol.EventUserLostConnect, action.EventOffline)
}
